package com.rainbow.support.half.message.handler.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.rainbow.common.core.toolkit.SpringBeanUtil;
import com.rainbow.support.half.message.entity.HalfMessagePayload;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.lang.NonNull;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;

import java.util.Objects;

/**
 * rocketMq半事务消费者
 * 通用逻辑块统一处理类
 * @author 傅为地
 */
@Slf4j
public abstract class AbstractConsumerHandler implements HalfMsgConsumerHandler {

    /**
     * TODO 消费者本地业务处理方法
     * 不同的消费者本地事务处理逻辑,
     * 具体处理自己本地事务逻辑
     * 本地业务处理时,消费人和消费时间最好
     * 采用消息体内的发送人和发送时间一致
     *
     * @param msg 消息主体
     * @return 消费者本地业务处理结果
     */
    public abstract boolean consumerProcess(@NonNull HalfMessagePayload msg);


    /**
     * 本地业务处理,开启事务处理
     * 事务回滚之后,异常继续上抛
     * 消费者捕获异常自行处理逻辑
     * @param producer 生产者参数
     * @param message  消费者消息
     * @return 本地事务处理结果
     */
    @Override
    public boolean serviceProcess(@NonNull HalfMessagePayload producer, @NonNull String message) {
        Boolean result = Objects.requireNonNull(SpringBeanUtil.getBean(TransactionTemplate.class), "事务组件初始化错误").execute(transactionStatus -> {
            try {
                saveConsumerLog(producer, message);
                consumerProcess(producer);
                return true;
            } catch (Exception error) {
                log.error("RocketMq消费者执行业务异常:", error);
                if (ObjectUtils.isNotEmpty(transactionStatus)) {
                    transactionStatus.setRollbackOnly();
                }
                throw error;
            }
        });
        return ObjectUtils.isEmpty(result) ? true : result;
    }

    /**
     * 消费者处理事务逻辑
     * 不推荐采用此方法处理消费者,消息体内日期格式可能存在问题
     * 推荐采用{@link #process(HalfMessagePayload producer)}
     * @param msg 消息主体
     * @return 消费者处理结果
     */
    @Deprecated
    @Override
    public boolean process(@NonNull String msg) {
        log.info("RocketMq半事务消费者,接收消息==>message:{}", msg);
        HalfMessagePayload source =parseHalfMessagePayload(msg);
        if (checkBranchId(source.getBranchId())) {
            log.warn("RocketMq半事务消费者,消息主体分支事务:" + source.getBranchId() + ",已成功消费.请勿重复消费!");
            return true;
        }
        prepareParams(source);
        serviceProcess(source, msg);
        return true;
    }


    /**
     * 消费者处理事务逻辑
     * @param source 消息主体
     * @return 消费者处理结果
     */
    @Override
    public boolean process(@NonNull HalfMessagePayload source) {
        String message = JSON.toJSONStringWithDateFormat(source, "yyyy-MM-dd HH:mm:ss", SerializerFeature.WriteMapNullValue);
        log.info("RocketMq半事务消费者,接收消息==>message:{}",message);
        Assert.notNull(source, "RocketMq半事务消费者,全局事务编号不能为空,请检查数据!");
        Assert.notNull(source.getGlobalId(), "RocketMq半事务消费者,全局事务编号不能为空,请检查数据!");
        Assert.notNull(source.getBranchId(), "RocketMq半事务消费者,分支事务编号不能为空,请检查数据!");
        Assert.notNull(source.getOriginBody(), "RocketMq半事务消费者,事务消息体不能为空,请检查数据!");
        Assert.notNull(source.getCreator(), "RocketMq半事务消费者,消息发送人不能为空,请检查数据!");
        Assert.notNull(source.getCreateTime(), "RocketMq半事务消费者,消息发送时间不能为空,请检查数据!");
        if (checkBranchId(source.getBranchId())) {
            log.warn("RocketMq半事务消费者,消息主体分支事务:" + source.getBranchId() + ",已成功消费.请勿重复消费!");
            return true;
        }
        prepareParams(source);
        serviceProcess(source, message);
        return true;
    }

}
